import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;

public class Test {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> source = env.readTextFile("data/a.txt");
        SingleOutputStreamOperator<Shop> map1 = source.map(new MapFunction<String, Shop>() {
            @Override
            public Shop map(String s) throws Exception {
                String[] sp = s.split(",");
                return new Shop(Integer.parseInt(sp[0]), sp[1], sp[2], Double.parseDouble(sp[3]), sp[4]);
            }
        }).assignTimestampsAndWatermarks(

                WatermarkStrategy
                        .<Shop>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((event, timestamp) -> {
                            try {
                                return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(event.getSj()).getTime();
                            } catch (ParseException e) {
                                throw new RuntimeException(e);
                            }
                        })

        );
        map1.print();
        SingleOutputStreamOperator<String> process = map1.keyBy(x -> "all").window(TumblingEventTimeWindows.of(Time.seconds(1)))
                .process(new ProcessWindowFunction<Shop, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<Shop, String, String, TimeWindow>.Context context, Iterable<Shop> iterable, Collector<String> collector) throws Exception {
                        int i = 0;
                        double b = 0.0;
                        for (Shop shop : iterable) {
                            i++;
                            b += shop.getPrice();
                        }
                        collector.collect("-zje-" + b + "-sl-" + i + "-pj-" + (b / i));
                    }
                });
        process.print();
//        SingleOutputStreamOperator<Tuple2<String, Double>> sum = map1.map(new MapFunction<Shop, Tuple2<String, Double>>() {
//            @Override
//            public Tuple2<String, Double> map(Shop shop) throws Exception {
//                return new Tuple2<>(shop.getLx(), shop.getPrice());
//            }
//        }).keyBy(0).sum(1);
//        sum.print();
        env.execute();
    }
}
